use rand::Rng;
use std::collections::HashMap;
use std::fs::File;
use std::io::{BufWriter, Error, Write};
use std::sync::Arc;
use tokio::sync::Mutex;

use crate::ast::{Command, Input, MqTopic, SocketState, Token};
use crate::{ast::Metadata, Connection};

#[derive(Debug)]
pub enum CommandState {
    CommandEnd,
    CommandConsumePanding,
    CommandProducePanding,
    CommandError,
}

#[derive(Debug)]
pub struct Handler {
    pub conn: Connection,
    pub db: Arc<Mutex<Metadata>>,
    pub topic: Option<String>,
    pub partition: Option<u16>,
    pub w_file: Option<HashMap<u16, BufWriter<File>>>,
    //pub rand: ThreadRng,
}

impl Handler {
    pub fn new(conn: Connection, db: Arc<Mutex<Metadata>>) -> Self {
        Handler {
            conn,
            db,
            topic: None,
            partition: None,
            w_file: None,
            //rand: rand::thread_rng(),
        }
    }

    pub async fn read_message(&mut self) -> Result<String, Error> {
        let line = self.conn.read_message().await?;
        println!("message: {}", line);

        if self.w_file.is_none() {
            let partition = self.partition.unwrap();
            let mut map = HashMap::new();
            let topic = self.topic.clone().unwrap_or("default".to_owned());
            for i in 0..partition {
                let file_name = format!("{}_{}.msg", topic, i);
                let file = File::create(file_name).unwrap();
                let writer = BufWriter::new(file);
                map.insert(i, writer);
            }
            self.w_file = Some(map);
        }

        let w_file_map = self.w_file.as_mut().unwrap();
        let mut rand = rand::thread_rng();
        let partition = rand.gen_range(0..self.partition.unwrap());

        if let Some(writer) = w_file_map.get_mut(&partition) {
            writer.write_all(line.as_bytes())?;
            writer.write_all(b"\n")?;
            writer.flush()?;
        }

        Ok(line)
    }

    pub async fn read_command(&mut self) -> Result<CommandState, Error> {
        let command_line = self.conn.read_message().await?;

        let buf = command_line.as_bytes();
        if buf.is_empty() {
            return Ok(CommandState::CommandError);
        }
        let mut input = Input::new(buf, buf.len() - 1);
        let analysis_option = input.analysis();
        if analysis_option.is_none() {
            self.conn
                .write_message("Please use the correct command")
                .await?;
            return Ok(CommandState::CommandError);
        }

        let token: Token = analysis_option.unwrap();

        match token.state {
            SocketState::CreateTopic => {
                let commad = Command::new(&token.args);
                let mq_topic = MqTopic::new(commad);
                {
                    let mut metadata_lock = self.db.lock().await;
                    metadata_lock.init_topic(mq_topic);
                }
                self.conn.write_message("run ok!\r\n").await?;
                return Ok(CommandState::CommandEnd);
            }
            SocketState::List => {
                {
                    let metadata_lock = self.db.lock().await;
                    match metadata_lock.get_all_topic() {
                        Some(topics) => {
                            let topic_str = format!("{:?} \r\n", topics);
                            self.conn.write_message(topic_str.as_str()).await?;
                        }
                        None => {
                            self.conn.write_message("no topic \r\n").await?;
                        }
                    }
                }
                return Ok(CommandState::CommandEnd);
            }
            SocketState::DeleteTopic => {
                let command = Command::new(&token.args);
                if let Some(command_topic) = command.topic {
                    let mut metadata_lock = self.db.lock().await;
                    metadata_lock.remove_topic(command_topic);
                }
                return Ok(CommandState::CommandEnd);
            }
            SocketState::Alter => {
                let command = Command::new(&token.args);
                if let (Some(command_topic), Some(command_partitions)) =
                    (command.topic, command.partitions)
                {
                    let mut metadata_lock = self.db.lock().await;
                    metadata_lock.change_partitions(command_topic, command_partitions);
                }
                return Ok(CommandState::CommandEnd);
            }
            SocketState::DeleteGroup => {}
            SocketState::Describe => {
                let command = Command::new(&token.args);
                if let Some(command_topic) = command.topic {
                    let mut metadata_lock = self.db.lock().await;
                    match metadata_lock.get_topic(command_topic) {
                        Some(mq_topic) => {
                            self.conn
                                .write_message(format!("{} \r\n", mq_topic).as_str())
                                .await?;
                        }
                        None => {
                            self.conn
                                .write_message("no topic, nothing to do \r\n")
                                .await?;
                        }
                    }
                }
                return Ok(CommandState::CommandEnd);
            }
            SocketState::Consume => {
                let command = Command::new(&token.args);
                if let (Some(command_topic), Some(command_group)) = (command.topic, command.group) {
                    let mut metadata_lock = self.db.lock().await;
                    metadata_lock.add_group(command_topic, command_group.to_owned());
                }
                return Ok(CommandState::CommandConsumePanding);
            }
            SocketState::Produce => {
                let command = Command::new(&token.args);
                if let Some(command_topic) = command.topic {
                    self.topic = Some(command_topic.to_owned());
                    let mut metadata_lock = self.db.lock().await;
                    if let Some(mq_topic) = metadata_lock.get_topic(command_topic) {
                        self.partition = Some(mq_topic.partitions);
                    }

                }

                return Ok(CommandState::CommandProducePanding);
            }
            SocketState::GetOffset => {}
            SocketState::None => {
                let _ = self.conn.write_message("run ok!\r\n").await;
                return Ok(CommandState::CommandEnd);
            }
        };
        Ok(CommandState::CommandEnd)
    }

    pub async fn write_message(&mut self, msg: &str) -> Result<(), Error> {
        self.conn.write_message(msg).await?;
        //let map = self.db.lock().unwrap();
        //match map.db.get(msg) {
        //    Some(msg) => {
        //        println!("msg: {}", msg);
        //    }
        //    None => {
        //        println!("none");
        //    }
        //}
        Ok(())
    }
}
